package com.atguigu.edu.realtime.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Properties;

/**
 * @author Lec
 * @date 2022/8/14 13:38
 */

public class KafkaUtil {

    private static String BOOTSTRAP_SERVERS = "hadoop102:9092,hadoop103:9092,hadoop104:9092";


    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic,String groupID) {
        Properties props = new Properties();

        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupID);

        //由于这里的SimpleStringSchema的deserialize有点小bug，所以要自定义一个

        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<String>(topic, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                if (record == null || record.value() == null) {
                    return "";
                }
                return new String(record.value());
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return BasicTypeInfo.STRING_TYPE_INFO;
            }
        }, props);

        return flinkKafkaConsumer;
    }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {
        Properties props = new Properties();

        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(topic, new SimpleStringSchema(), props);

        return kafkaProducer;

    }

    public static String getKafkaDDL(String topicName,String groupID) {

        return  " WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '"+topicName+"',\n" +
                "  'properties.bootstrap.servers' = '"+ BOOTSTRAP_SERVERS +"',\n" +
                "  'properties.group.id' = '"+groupID+"',\n" +
                "  'scan.startup.mode' = 'group-offsets',\n" +
                "  'format' = 'json'\n" +
                ")";
    }
    public static String getKafkaSinkDDL(String topicName){
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '"  + topicName + "',\n" +
                "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "',\n" +
//                "  'key.format' = 'json' , \n" +
                "  'value.format' = 'json'  \n" +
                ")";
    }
    public static String getUpsertKafkaSinkDDL(String topicName){
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '"  + topicName + "',\n" +
                "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVERS + "',\n" +
                "  'key.format' = 'json' , \n" +
                "  'value.format' = 'json'  \n" +
                ")";
    }
}
